#include "iothreadpool.h"
#include "server.h"

IOThreadPool::IOThreadPool()
{
    io_thread_op_ = IO_THREADS_OP_READ;
}

IOThreadPool::~IOThreadPool()
{}

int IOThreadPool::getIOPendingCount(int i)
{
    if(i >= IO_THREADS_MAX_NUM)
        return 0;
    //std::cout << "getIOPendingCount: " << io_threads_pending[i] << std::endl;
    return io_threads_pending[i];
}

void IOThreadPool::setIOPendingCount(int i, int count)
{
    if(i >= IO_THREADS_MAX_NUM)
        return;
    io_threads_pending[i] = count;
}


bool IOThreadPool::postponeClientRead(connection* conn)
{
    if(server.io_thread_active_ && !(conn->flags_ & CONN_PENDING_READ))
    {
        conn->flags_ = CONN_PENDING_READ;
        server.clients_pending_read_.push_back(conn);
    }
    else 
    {
        return false;
    }
}

int IOThreadPool::readQueryFromClient(connection* conn)
{
    //判断是否加入io多线程，读全局队列
    if(server.io_thread_->postponeClientRead(conn))
        return 0;
    //std::cout << "readQueryFromClient " << conn << ":" << pthread_self() << std::endl;
    int querybuf_len = strlen(conn->query_buf_);
    int readlen = read(conn->fd_, conn->query_buf_ + querybuf_len, IO_QUERY_BUFF_MAX_NUM - querybuf_len);
    if(readlen == 0)
    {
        std::cout << "readQueryFromClient delete: " << conn << ":" << pthread_self() << "fd: " << conn->fd_ << std::endl;
        //client close
        server.freeClientAsync(conn);
        return -1;
    }
    else if(readlen < 0)
    {
        if(errno == EAGAIN)
            return 0;
        //read error
        std::cout << "readQueryFromClient delete2: " << conn << ":" << pthread_self() << "fd: " << conn->fd_  << "errno: " << errno << std::endl;
        server.freeClientAsync(conn);
        return -1;
    }

    conn->query_buf_[querybuf_len + readlen] = '\0';
    conn->processInputBuffer();

    //加入到写全局队列
    if(!(conn->flags_ & CONN_PENDING_WRITE) || conn->clientHasPendingReplies())
        server.clientInstallWriteHandler(conn);

    return 0;
}

int IOThreadPool::writeToClient(connection* conn, int handle_install)
{
    while(conn->hasPendingReplies())
    {
        if(conn->buf_pos_ > 0)
        {
            int nwritten = write(conn->fd_, conn->buf_ + conn->sendlen_, conn->buf_pos_ - conn->sendlen_);
            if(nwritten < 0)
            {
                if(errno == EAGAIN)
                    continue;
                //std::cout << "writeToClient delete: " << conn << std::endl;
                server.freeClientAsync(conn);
                return -1;
            }
            else if(nwritten == 0)
            {
                //std::cout << "writeToClient delete2: " << conn << std::endl;
                // close(conn->fd_);
                // delete conn;
                break;
            }
    
            conn->sendlen_ += nwritten;
            if(conn->buf_pos_ == conn->sendlen_)
            {
                conn->buf_pos_ = 0;
                conn->sendlen_ = 0;
            }
        }
        else
        {
            //reply节点
            replyBlock* o = conn->reply_.front();
            size_t objlen = o->used_;
            if(objlen == 0)
            {
                //没有有效数据
                conn->reply_size_ -= o->size_;
                conn->reply_.pop_front();
                continue;
            }

            int nwritten = write(conn->fd_, &o->buf_[conn->sendlen_], o->used_ - conn->sendlen_);
            if(nwritten < 0)
            {
                //std::cout << "writeToClient delete: " << conn << "errno: " << errno << std::endl;
                if(errno == EAGAIN)
                    continue;
                server.freeClientAsync(conn);
                return -1;
            }
            else if(nwritten == 0)
            {
                break;
            }

            conn->sendlen_ += nwritten;
            if(objlen == conn->sendlen_)
            {
                conn->reply_size_ -= o->size_;
                conn->reply_.pop_front();
                conn->sendlen_ = 0;
                //free(o);
                //operator delete(o);
                // o->~replyBlock();
                // cFree(o);
                tcfreeSize(replyBlock, o);

                if(conn->reply_.empty())
                    conn->reply_size_ = 0;
            }
        }

    }

    if(!conn->hasPendingReplies())
    {
        if(handle_install)
            conn->connnSocketSetWriteHandler(nullptr);
    }
    //close(conn->fd_);
    //delete conn;
    return 0;
}

int IOThreadPool::sendReplyToClient(connection* conn)
{
    writeToClient(conn, 1);
}

void* IOThreadPool::IOThreadMain(void* arg)
{
    long id = (unsigned long)arg;
    while(1)
    {
        /* Wait for start */
        for (int j = 0; j < 1000000; j++) {
            if (server.io_thread_->getIOPendingCount(id) != 0) break;
        }

        if(server.io_thread_->getIOPendingCount(id) == 0)
        {
            //std::cout << "pthread_mutex_lock: " << pthread_self() << "id: " << id << std::endl;
            server.io_thread_->waitIOThread(id);
            //std::cout << "pthread_mutex_unlock: " << pthread_self() << "id: " << id << std::endl;
        }
        //处理队列中的任务，一次性处理完写任务或者读任务
        server.io_thread_->handleTask(id);
        server.io_thread_->setIOPendingCount(id, 0);
    }
}

void IOThreadPool::InitThreadIO()
{
    if(server.io_thread_num_ == 1)return;

    if(server.io_thread_num_ > IO_THREADS_MAX_NUM)
    {
        assert(server.io_thread_num_ <= IO_THREADS_MAX_NUM);
    }

    for(int i = 0; i < server.io_thread_num_; ++i)
    {
        //i==0是主线程
        if(i == 0)continue;

        pthread_mutex_init(&io_thread_mutex_[i], NULL);
        io_threads_pending[i] = 0;
        //主线程获得锁，让io线程阻塞
        pthread_mutex_lock(&io_thread_mutex_[i]);
        pthread_t tid;
        if(pthread_create(&tid, NULL, IOThreadMain, (void*)(long)i))
        {
            exit(1);
        }
        io_threads[i] = tid;
    }
}

void IOThreadPool::startThreadIO()
{
    for(int i = 1; i < server.io_thread_num_; ++i)
    {
        pthread_mutex_unlock(&io_thread_mutex_[i]);
    }
    server.io_thread_active_ = 1;
}

void IOThreadPool::stopThreadIO()
{
    //server.handleClientsWithPendingReadsUsingThreads();
    server.handleClientsWithPendingReadsUsingThreads();
    for(int i = 1; i < server.io_thread_num_; ++i)
    {
        pthread_mutex_lock(&io_thread_mutex_[i]);
    }
    server.io_thread_active_ = 0;
}

void IOThreadPool::pushThreadList(int i, connection* conn)
{
    if(i >= IO_THREADS_MAX_NUM || i < 0)
        return;
    io_thread_list_[i].push_back(conn);
}

int IOThreadPool::getThreadListSize(int i)
{
    if(i >= IO_THREADS_MAX_NUM || i < 0)
        return 0;
    return io_thread_list_[i].size();
}

void IOThreadPool::setThreadOperation(int op)
{
    io_thread_op_ = op;
}


void IOThreadPool::handleReadTask(int i)
{
    //主线程处理任务
    std::list<connection*>::iterator io_it = io_thread_list_[i].begin();
    for(; io_it != io_thread_list_[i].end(); ++io_it)
    {
        readQueryFromClient(*io_it);
    }
    io_thread_list_[i].clear();
}

void IOThreadPool::handleWriteTask(int i)
{
    //主线程处理任务
    std::list<connection*>::iterator io_it = io_thread_list_[i].begin();
    for(; io_it != io_thread_list_[i].end(); ++io_it)
    {
        writeToClient(*io_it, 0);
    }
    io_thread_list_[i].clear();
}

void IOThreadPool::handleTask(int i)
{
    //处理队列中的任务，一次性处理完写任务或者读任务
    std::list<connection*>::iterator it = io_thread_list_[i].begin();
    for(; it != io_thread_list_[i].end(); ++it)
    {
        connection* conn = *it;
        if(io_thread_op_ == IO_THREADS_OP_READ)
        {
            //std::cout << "readQueryFromClient pthread_self: " << pthread_self() << " id: " << i << std::endl;
            readQueryFromClient(conn);
        }
        if(io_thread_op_ == IO_THREADS_OP_WRITE)
        {
            //std::cout << "writeToClient pthread_self: " << pthread_self() << " id: " << i << std::endl;
            writeToClient(conn);
        }
        
    }
    io_thread_list_[i].clear();
}

void IOThreadPool::waitIOThread(int i)
{
    if(i >= IO_THREADS_MAX_NUM || i < 0)
        return;
    
    pthread_mutex_lock(&io_thread_mutex_[i]);
    pthread_mutex_unlock(&io_thread_mutex_[i]);
}